package hbase.DelFriends;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;


//重写prePut方法，监听到向lagou_friends表插入数据时，执行向lagou_friends表插入数据的代码
public class MyFriendsProcessor extends BaseRegionObserver {

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //删除好友信息
        //获取被删除的对象信息
        final HTableInterface lagou_friends = e.getEnvironment().getTable(TableName.valueOf("lagou_friends"));
        //解析lagou_friends表的删除对象

        NavigableMap<byte[], List<Cell>> delteCellMap =  delete.getFamilyCellMap();
        //通过队列集合
        Set<Map.Entry<byte[], List<Cell>>> entries = delteCellMap.entrySet();
        //遍历数组执行删除
        for (final Map.Entry<byte[], List<Cell>> entry : entries) {
            List<Cell> cells = entry.getValue();
            //循环cells对象获取每个cell的rowkey以及列族
            for (final Cell cell : cells) {
                byte[] rowkey = CellUtil.cloneRow(cell);
                System.out.println("delete.rowkey ---- " + rowkey);
                byte[] qualifier = CellUtil.cloneQualifier(cell);
                System.out.println("delete.qualifier ---- " + qualifier);
                //判断是否存在删除的rowkey的cell
                boolean flag = lagou_friends.exists(new Get(qualifier).addColumn(Bytes.toBytes("friends"), rowkey));
                if (flag){
                    //生成一个delete对象对需要删除的列族进行删除
                    Delete delQualifier = new Delete(qualifier).addColumn(Bytes.toBytes("friends"), rowkey);
                    lagou_friends.delete(delQualifier);
                }
        //关闭table对象
        lagou_friends.close();
            }
        }
    }
}